package DianShang_2024.ds_02.tzgc


import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.functions.{col, desc, sum, when}

import java.util.Properties

object tzgc01 {
  def main(args: Array[String]): Unit = {
    /*
          剔除订单信息表与订单详细信息表中用户id与商品id不存在现有的维表中的记录，同时建议多利用缓存并充分考虑并行度来优化代
          码，达到更快的计算效果。

    1、根据Hudi的dwd库中相关表或MySQL中shtd_store中相关表（order_detail、sku_info），计算出与用户id为6708的用户所购
    买相同商品种类最多的前10位用户（只考虑他俩购买过多少种相同的商品，不考虑相同的商品买了多少次），将10位用户id进行输出，输
    出格式如下，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下；
    结果格式如下：
    -------------------相同种类前10的id结果展示为：--------------------
    1,2,901,4,5,21,32,91,14,52
     */

    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("第二套卷子的特征工程第一题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    val order_info=spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_info",connect)
    val order_detail=spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_detail",connect)

    //  为了解决dataframe转化为dataset的数据类型报错，导入spark的隐式转换
    import spark.implicits._


    //  拿到所有用户和其购买商品信息
    val all_user_buy_sku=order_info.join(order_detail,order_detail("order_id")===order_info("id"))
      .select("user_id","sku_id")
      .distinct()       //  因为不考虑用户购买了多少次，所以这里进行去重


    //  拿到用户6708购买的商品id
    val user_6708_buy_ids:Array[Int]=all_user_buy_sku
      .filter(col("user_id")===2790)
      .select("sku_id")
      .map(r => r(0).toString.toInt)
      .collect()

    // 拿到用户购买商品id和6708购买种类相似度最高的
    //  添加一个字段判断用户购买的商品种类，6708是否买过，买过为1，没有为0，然后根据这个字段的值拿到前10的id
    val result:Array[String]=all_user_buy_sku
      .filter(col("user_id")!==2790)
      .withColumn(
        "pd",
        when(col("sku_id").isin(user_6708_buy_ids:_*),1).otherwise(0)
      )
      .groupBy("user_id")
      .agg(sum("pd").as("count_pd"))
      .orderBy(desc("count_pd"))
      .limit(10)
      .select("user_id")
      .map(r => r(0).toString)
      .collect()
println("---------------------------相同种类前10的用户id----------------------------")
    println(result.mkString(","))



    spark.close()
  }

}
